package cn.doitedu.rtmk.demo2;


import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.math.BigDecimal;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/22
 * @Desc: 学大数据，上多易教育
 * 规则：
 * 静态画像条件： age>30岁，月消平均费额>300的用户
 * 动态画像条件： 在规则上线后，发生 X行为 >= 3次，发生  (A * W * E) 行为序列 >= 2次
 * 触发事件：    当他做 add_cart 行为时，给他推送消息
 **/
public class Demo2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);


        // 创建kafka 行为明细topic的source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doit43-8")
                .setClientIdPrefix("doitedu-c")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<UserEvent> eventStream = streamSource.map(json -> JSON.parseObject(json, UserEvent.class));

        SingleOutputStreamOperator<String> hitMessages = eventStream
                .keyBy(UserEvent::getUser_id)
                .process(new KeyedProcessFunction<Long, UserEvent, String>() {
                    ValueState<Integer> xCntState;
                    ValueState<Integer> idxState;
                    ValueState<Integer> seqCntState;

                    String[] eventSeq = {"A", "W", "E"};

                    // hbase的表连接
                    Table userProfileTable;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // x动态画像标签的统计计数器
                        xCntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("x_cnt", Integer.class));

                        // AWE序列完成到了第几步的计数器
                        idxState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("idx", Integer.class));

                        // AWE序列完整完成的次数计数器
                        seqCntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("seq_cnt", Integer.class));


                        // 创建一个hbase的查询客户端
                        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
                        hbaseConf.set("hbase.zookeeper.quorum", "doitedu:2181");
                        Connection hbaseConn = ConnectionFactory.createConnection(hbaseConf);
                        userProfileTable = hbaseConn.getTable(TableName.valueOf("user_profile"));

                    }

                    @Override
                    public void processElement(UserEvent userEvent, KeyedProcessFunction<Long, UserEvent, String>.Context ctx, Collector<String> out) throws Exception {

                        String eventId = userEvent.getEvent_id();
                        long userId = userEvent.getUser_id();

                        // 先判断收到的用户行为，是否是 规则的动态画像条件1所需要统计的行为X
                        if ("X".equals(eventId)) {
                            xCntState.update(xCntState.value() == null ? 1 : xCntState.value() + 1);
                        }


                        // 再判断收到的用户行为，是否是 规则的动态画像条件2所需要统计的行为 (A W E) >= 2次
                        int idx = idxState.value() == null ? 0 : idxState.value();
                        // 判断收到的用户行为，是否是动态画像行为序列统计条件中正在等待的事件
                        if (eventId.equals(eventSeq[idx])) {

                            if (idx == 2) {
                                // 如果当前的idx是2，说明已经是等最后一个事件，而此时进来的正好是等待的最后一个事件
                                // 则将idx重置为0，而将序列的整体完成次数+1
                                idxState.update(0);
                                seqCntState.update(seqCntState.value() == null ? 1 : seqCntState.value() + 1);
                            } else {
                                idxState.update(idx + 1);
                            }
                        }


                        // 再判断收到的用户行为，是否是 规则中的触发事件
                        if (eventId.equals("add_cart")) {

                            // 如果是，则判断该用户的动态画像条件是否已经满足
                            Integer xCnt = xCntState.value();
                            Integer seqCnt = seqCntState.value();
                            if (xCnt != null && xCnt >= 3 && seqCnt != null && seqCnt >= 2) {
                                // 如果动态画像条件以满足，则继续判断该用户的静态画像条件是否也满足
                                Get get = new Get(Bytes.toBytes(userId+""));
                                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("tag102"));

                                Result result = userProfileTable.get(get);
                                byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                                String userAgeStr = Bytes.toString(ageBytes);
                                int userAge = Integer.parseInt(userAgeStr);

                                byte[] tag102Bytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("tag102"));
                                String tag102Str = Bytes.toString(tag102Bytes);
                                double tag102 = Double.parseDouble(tag102Str);

                                if (userAge > 30 && tag102 > 300) {
                                    // 如果也满足，则输出规则命中消息
                                    JSONObject message = new JSONObject();
                                    message.put("user_id", userId);
                                    message.put("hit_time", userEvent.getAction_time());
                                    message.put("rule_id", "rule-001");

                                    out.collect(message.toJSONString());

                                }


                            }


                        }
                    }
                });


        hitMessages.print();


        env.execute();

    }


}
